# -*- coding:utf-8 -*-
#author: mz.mail@foxmail.com
#edition: V1.0

from toollib.guid import SnowFlake
import urllib.parse as urlp
import alive_progress #1.6.2
import pandas as pd
import numpy as np
import sqlalchemy
import traceback
import cx_Oracle
import sqlparse
import pymysql
import uuid
import time
import re
import os

#定义常量
dbtype=1;debug=0;dbinfo={'host':"",'user':"",'password':"",'database':"",'port':1521}

#获取去掉连字符的UUID
def UUID():
    rs = str(uuid.uuid1())
    rs = rs.replace("-","")
    return rs

#获取SnowID
def SnowID():
    Snow = SnowFlake()
    rs = str(Snow.gen_uid())
    return rs

#读取excel文件到dataframe对象
def readxl(s):
    try:
        ptn1=re.compile(r'(.*)\.xls([x]{0,1})', re.I)
        ptn2=re.compile(r'(.*)\.csv', re.I)
        pd.set_option('display.unicode.east_asian_width',True)
        xl=getinput2(s,ptn1,ptn2)
        if re.fullmatch(ptn1,xl) is not None:
            df=pd.read_excel("./"+str(xl),0)
            print("EXCEL文件读取成功...");print(df.head())
        else:
            df=pd.read_csv("./"+str(xl))
            print("CSV文件读取成功...");print(df.head())
        return df
    except:
        print(str(os.sys.exc_info()[1]))
        print(traceback.format_exc())
        print("EXCEL文件读取失败!")
        return None

#获取输入信息
def getinput(txt,pattern):
    print(txt)
    ipt = input()
    while re.fullmatch(pattern,ipt) is None:
        print("输入错误,请重新输入:")
        ipt = input()
    return str(ipt)

#获取输入信息2
def getinput2(txt,pattern1,pattern2):
    print(txt)
    ipt = input()
    while re.fullmatch(pattern1,ipt) is None and re.fullmatch(pattern2,ipt) is None:
        print("输入错误,请重新输入:")
        ipt = input()
    return str(ipt)

#打印日志
def printlog(txt):
    try:
        print(txt)
        filepath = os.getcwd()
        now = str(time.strftime("%Y-%m-%d %H:%M:%S",time.localtime()))
        date=str(time.strftime("%Y_%m_%d",time.localtime()))
        with open(filepath + "/log_ieid_"+date+".log","a",encoding='utf-8') as log:
            log.write(now + " " + str(txt) + '\n')
    except:
        print(str(os.sys.exc_info()[1]))
        print(traceback.format_exc())
        print("写入日志失败...")

#测试数据库链接
def dbconntest(ptconn=False):
    try:
        global dbinfo;global dbtype;hoststr=dbinfo["host"]
        dbinfo=getport(dbinfo);hoststr=dbinfo["host"]+":"+dbinfo["port"]+"/"+dbinfo["database"]
        try:
            db=cx_Oracle.connect(dbinfo["user"],dbinfo["password"],dbinfo["host"]+":"+dbinfo["port"]+"/"+dbinfo["database"]);dbtype=0;db.close()
            dbinfo["conn"]='oracle+cx_oracle://'+dbinfo["user"]+":"+dbinfo["password"]+"@"+dbinfo["host"]+":"+dbinfo["port"]+"/"+dbinfo["database"]
            print("已连接Oracle:"+dbinfo["user"]+"/"+dbinfo["password"]+"@"+dbinfo["host"]+":"+dbinfo["port"]+"/"+dbinfo["database"])
        except:
            try:
                import dmPython;db=dmPython.connect(user=dbinfo["user"],password=dbinfo["password"],server=dbinfo["host"],port=int(dbinfo["port"]));dbtype=2;db.close()
                dbinfo["conn"]='dm+dmPython://'+dbinfo["user"]+":"+urlp.quote_plus(dbinfo["password"])+"@"+dbinfo["host"]+":"+dbinfo["port"]
                print("已连接DAMENG:"+dbinfo["user"]+"/"+dbinfo["password"]+"@"+dbinfo["host"]+":"+dbinfo["port"]+"/"+dbinfo["database"])
            except:
                db=pymysql.connect(host=dbinfo["host"],user=dbinfo["user"],password=dbinfo["password"],database=dbinfo["database"],port=int(dbinfo["port"]));dbtype=1;db.close()
                dbinfo["conn"]='mysql+pymysql://'+dbinfo["user"]+":"+urlp.quote_plus(dbinfo["password"])+"@"+dbinfo["host"]+":"+dbinfo["port"]+"/"+dbinfo["database"]
                print("已连接MySQL:"+dbinfo["user"]+"/"+dbinfo["password"]+"@"+dbinfo["host"]+":"+dbinfo["port"]+"/"+dbinfo["database"])
        return 1
    except:
        print("数据库服务器连接失败!请检查配置后重试...")
        try:
            print("当前配置如下:\r\nUSERID="+dbinfo["user"]+";PASSWD="+dbinfo["password"]+";DBLINK="+hoststr)
            if str(dbinfo["password"]).startswith(' '):print('注意: 当前输入密码开头有空格!')
            if str(dbinfo["password"]).endswith(' '):print('注意: 当前输入密码末尾有空格!')
        except:
            pass
            return 9

#解析端口
def getport(infod:dict):
    try:
        info=infod
        host=info["host"]
        pos_port=host.find(":")
        pos_databs=host.find("/")
        if pos_databs==-1:
            return info
        elif pos_port==-1:
            info["host"]=host[:pos_databs]
            info["port"]="3306"
            info["database"]=host[pos_databs+1:]
        else:
            info["host"]=host[:pos_port]
            info["port"]=host[pos_port+1:pos_databs]
            info["database"]=host[pos_databs+1:]
        return info
    except:
        print(str(os.sys.exc_info()[1]))
        print(traceback.format_exc())
        print("MySQL端口识别失败,请完整填写IP及数据库名,例如,127.0.0.1:3306/audit,默认3306端口时,可省略端口配置!")
        return dbinfo

#测试SQLalchemy连接池是否可用
def sqlalchemytest(caution=False):
    try:
        sqlsentence="SELECT 1 FROM DUAL"
        pool=sqlalchemy.create_engine(dbinfo["conn"],pool_recycle=3600,echo=False,pool_size=10,max_overflow=1000)
        conn=pool.connect()
        cur=conn.execute(sqlalchemy.text(sqlsentence))
        cur.fetchone()
        conn.close()
        pool.dispose()
        return 1
    except:
        if caution==True:print("CAUTION: SQLALCHEMY CONNECTION POOL IS NOT SUPPORTED!")
        return 0
    
#建立数据库连接池
def makedbconn():
    try:
        if dbinfo['sqlalchemypool']==1:
            pool=sqlalchemy.create_engine(dbinfo["conn"],pool_recycle=3600,echo=False,pool_size=10,max_overflow=1000)
        else:
            pool=None
        return pool
    except:
        print(str(os.sys.exc_info()[1]))
        print(traceback.format_exc())
        return None

#从连接池建立单次使用的数据库链接
def getsigconn(conn):
    if conn is not None:
        sigconn=conn.raw_connection()
    else:
        if dbtype==0:#oracle
            sigconn=cx_Oracle.connect(dbinfo["user"],dbinfo["password"],dbinfo["host"]+":"+dbinfo["port"]+"/"+dbinfo["database"])
        elif dbtype==1:#mysql
            sigconn=pymysql.connect(host=dbinfo["host"],user=dbinfo["user"],password=dbinfo["password"],database=dbinfo["databs"],port=int(dbinfo["port"]),connect_timeout=1000)
        elif dbtype==2:#dameng
            import dmPython;sigconn=dmPython.connect(user=dbinfo["user"],password=dbinfo["password"],server=dbinfo["host"],port=int(dbinfo["port"]))
    return sigconn

#断开数据库
def disposedbconn(conn):
    try:
        if conn is not None:
            conn.dispose()
        else:
            pass
    except:
        print(str(os.sys.exc_info()[1]))
        print(traceback.format_exc())

#查询并返回第一条结果
def sqlqueryone(conn,sqlsentence):
    try:
        sqlsentence=sqlparse.format(sqlsentence,encoding=None,reindent=True, strip_comments=True,keyword_case='upper',identifier_case='upper')
        if debug==1:print(sqlsentence)
        conn=getsigconn(conn)
        cur=conn.cursor()
        cur.execute(sqlsentence)
        rs = cur.fetchone()
        cur.close()
        conn.close()
        if debug==1:print('DATA QUERY: '+str(len(rs))+' ROWS!')
        return rs
    except:
        print(sqlsentence)
        print(str(os.sys.exc_info()[1]))
        print(traceback.format_exc())
        print("数据库查询错误!")
        return None

#查询并返回所有结果
def sqlqueryall(conn,sqlsentence):
    try:
        sqlsentence=sqlparse.format(sqlsentence,encoding=None,reindent=True, strip_comments=True,keyword_case='upper',identifier_case='upper')
        if debug==1:print(sqlsentence)
        conn=getsigconn(conn)
        cur=conn.cursor()
        cur.execute(sqlsentence)
        rs = cur.fetchall()
        cur.close()
        conn.close()
        if debug==1:print('DATA QUERY: '+str(len(rs))+' ROWS!')
        return rs
    except:
        print(sqlsentence)
        print(str(os.sys.exc_info()[1]))
        print(traceback.format_exc())
        print("数据库查询错误!")
        return None

#查询并返回所有结果并返回列名
def sqlqueryallwithcols(conn,sqlsentence,trace=True):
    try:
        sqlsentence=sqlparse.format(sqlsentence,encoding=None,reindent=True, strip_comments=True,keyword_case='upper',identifier_case='upper')
        if debug==1:print(sqlsentence)
        conn=getsigconn(conn)
        cur=conn.cursor()
        cur.execute(sqlsentence)
        rs = cur.fetchall()
        cols = [str(i[0]).upper() for i in cur.description]
        cur.close()
        conn.close()
        if debug==1:print('DATA QUERY: '+str(len(rs))+' ROWS!')
        return rs,cols
    except:
        if trace==True:
            print(sqlsentence)
            print(str(os.sys.exc_info()[1]))
            print(traceback.format_exc())
            print("数据库查询错误!")
        return None,None

 #查询并返回所有结果并返回列名,并转换为DataFrame
def sqlqueryall2dataframe(conn,sqlsentence):
    try:
        rs,cols=sqlqueryallwithcols(conn,sqlsentence)
        rspd=pd.DataFrame(list(rs),columns=cols)
        if debug==1:print(rspd.head())
        return rspd
    except:
        print(str(os.sys.exc_info()[1]))
        print(traceback.format_exc())
        print("DataFrame转换错误!")
        return None

#执行一条sql语句
def sqlexcute(conn,sqlsentence,trace=True):
    try:
        sqlsentence=sqlparse.format(sqlsentence,encoding=None,reindent=True, strip_comments=True,keyword_case='upper',identifier_case='upper')
        if debug==1:print(sqlsentence)
        conn=getsigconn(conn)
        cur=conn.cursor()
        cur.execute(sqlsentence)
        conn.commit()
        cur.close()
        conn.close()
    except:
        if trace==True:
            print(sqlsentence)
            print(str(os.sys.exc_info()[1]))
            print(traceback.format_exc())
            print("数据库写入错误!")

#执行批量插入 sqlinsertlist
def sqlinsertlist(conn,sqlsentence,sqllist):
    try:
        sqlsentence=sqlparse.format(sqlsentence,encoding=None,reindent=True, strip_comments=True,keyword_case='upper',identifier_case='upper')
        if debug==1:print(sqlsentence);print(sqllist)
        if dbtype in [0,2]:
            conn=getsigconn(conn)
            cursor=conn.cursor()
            cursor.executemany(sqlsentence,sqllist)
            conn.commit()
            cursor.close()
            conn.close()
        elif dbtype==1:
            if conn is not None:
                sqlsentence=str(sqlsentence).replace(' ','')
                pattern=re.compile(r'insert(\s{0,})into(.*?)[(](.*?)[)](\s{0,})values',re.I) 
                sqlhead=pattern.match(sqlsentence).groups()
                tablename=sqlhead[1]
                tablecols=sqlhead[2].split(',')
                df_data=pd.DataFrame(sqllist,columns=tablecols)
                if dbinfo['mysql_lower_case_table_names']==1:tablename=tablename.lower()
                df_data.to_sql(tablename,conn,if_exists='append',index=False,chunksize=100,method='multi')
            else:
                conn=getsigconn(conn)
                cur=conn.cursor()
                pattern=re.compile(r'insert(\s)into(.*)[(](.*)[)](\s{0,})values(\s{0,})[(]',re.I) 
                sqlhead=pattern.match(sqlsentence).group()
                if len(sqllist)<10:
                    for sqls in sqllist:
                        newsql=sqlhead
                        for val in sqls:
                            if val is None or pd.isna(val):
                                newsql=newsql+"NULL,"
                            else:
                                newsql=newsql+"'"+str(val).replace("'","''")+"',"
                        newsql=newsql.rstrip(",")+")"
                        #if debug==1:print(newsql)
                        cur.execute(newsql)
                else:
                    with alive_progress.alive_bar(len(sqllist)) as bar:
                        for sqls in sqllist:
                            newsql = sqlhead
                            for val in sqls:
                                if val is None or pd.isna(val):
                                    newsql=newsql+"NULL,"
                                else:
                                    newsql=newsql+"'"+str(val).replace("'","''")+"',"
                            newsql=newsql.rstrip(",")+")"
                            #if debug==1:print(newsql)
                            cur.execute(newsql)
                            bar()
                conn.commit()
                cur.close()
                conn.close()
    except:
        printlog(str(os.sys.exc_info()[1]))
        printlog(traceback.format_exc())
        print("数据库写入错误!")
    
#测试用户数据库连接
def userconnectiontest():
    try:
        print("测试数据库连接...");global dbinfo;global dbtype
        def getnewlink():
            dbinfo["host"]=getinput("请输入数据库链接,例如,[192.168.10.10:33306/mydatabase]:",re.compile(r'(.*)', re.I)).strip()
            dbinfo["user"]=getinput("请输入数据库用户名,例如,[myaccount]:",re.compile(r'(.*)', re.I)).strip()
            dbinfo["password"]=getinput("请输入数据库用户密码,例如,[mypassword]:",re.compile(r'(.*)', re.I))
            while dbconntest() not in (1,2):
                dbinfo["host"]=getinput("请输入数据库链接,例如,[192.168.10.10:33306/mydatabase]:",re.compile(r'(.*)', re.I)).strip()
                dbinfo["user"]=getinput("请输入数据库用户名,例如,[myaccount]:",re.compile(r'(.*)', re.I)).strip()
                dbinfo["password"]=getinput("请输入数据库用户密码,例如,[mypassword]:",re.compile(r'(.*)', re.I))
        if dbinfo['host'] !="":
            print("当前数据库配置为:\r\nUSERID="+dbinfo["user"]+";PASSWD="+dbinfo["password"]+";DBLINK="+dbinfo["host"]+":"+str(dbinfo["port"])+"/"+dbinfo["database"])
            if getinput("是否使用当前配置? y/n:",re.compile(r'([yn]{1})', re.I)).lower()=='y':
                pass
            else:
                getnewlink()
        else:
            getnewlink()
        dbinfo['sqlalchemypool']=sqlalchemytest(True)#测试sqlalchemy连接池是否可用
        conn=makedbconn()
        if dbtype==1:dbinfo['mysql_lower_case_table_names']=int(sqlqueryone(conn,"SHOW GLOBAL VARIABLES like 'lower_case_table_names'")[1])
        disposedbconn(conn)#关闭数据库链接池
        return 1
    except:
        print(str(os.sys.exc_info()[1]))
        print(traceback.format_exc())
        print("数据库服务器连接失败!请检查配置重试...")
        return 0

#print df head
def printdfheadtranspose(df:pd.DataFrame):
    if len(df.columns)<=6:
        print(df.head())
    if len(df.columns)>6:
        cols=['列名','第1行','第2行','第3行'];rows=3
        if len(df)<3:cols=cols[:len(df)+1];rows=len(df)
        dfhv=pd.DataFrame(np.array(df.head(rows).transpose().reset_index()).tolist(),columns=cols)
        print(dfhv)